package com;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import org.jsoup.nodes.Element;
import org.jsoup.select.Elements;

public class BooksData {

	private static int books = 0;
	private static int hbooks = 0;
	private static int co = 0;
	private static List<String> url = new ArrayList<String>();
	private static List<String> inf = new ArrayList<String>();
	private static List<String> bs = new ArrayList<String>();
	private static Configuration config = null;
	/**
	 * 初始化配置
	 */
	static {
		config = HBaseConfiguration.create();
	}

	public static void main(String[] args) {
		String tableName = "books";
		String[] familys = { "info", "store" };

		try {
			createTable(tableName, familys);
			Job job = Job.getInstance(config, "books");
			job.setJarByClass(BooksData.class); // class that contains mapper
			// and
			// reducer

			Scan scan = new Scan();
			scan.setCaching(500); // 1 is the default in Scan, which will be
									// bad
									// for MapReduce jobs
			scan.setCacheBlocks(false); // don't set to true for MR jobs
			scan.setBatch(1000);
			// set other scan attrs
			// scan.addColumn(family, qualifier);
			TableMapReduceUtil.initTableMapperJob("crawldb_webpage", // input
																		// table
					scan, // Scan instance to control CF and attribute
							// selection
					MyMapper.class, // mapper class
					Text.class, // mapper output key
					Text.class, // mapper output value
					job);
			TableMapReduceUtil.initTableReducerJob(tableName, // output
					// table
					MyTableReducer.class, // reducer class
					job);
			job.setNumReduceTasks(1); // at least one, adjust as required

			job.waitForCompletion(true);
			System.out.println("==========END========");
			viewTable(tableName);
			System.out.println();
			System.out.println("书量" + ":" + books + "/" + (hbooks + co));
			System.out.println("多址");
			System.out.println(url);
			System.out.println("多书");
			System.out.println(bs);
			System.out.println("找茬");
			System.out.println(inf);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public static void viewTable(String tableName) throws IOException {
		HTable htable = new HTable(config, tableName);
		Scan scan = new Scan();
		ResultScanner scanner = htable.getScanner(scan);
		String regEx = "[\\w\u00C0-\uFFE5  -;\\[\\]·?<>×~@{}]*";
		for (Result r : scanner) {
			byte[] data = r.getRow();
			int offset = 0;
			int length = 0;
			String charset = "UTF-8";
			String datas = new String(data, charset);
			
			boolean matcher = datas.matches(regEx);
			if (!matcher) {
				for (int i = 0; i < data.length; i++) {
			//		System.out.print(data[i]);
				}
				System.out.println();
				System.out.println(datas);
			}

			++hbooks;
			Cell[] rc = r.rawCells();
			for (Cell cell : rc) {
				data = cell.getFamilyArray();
				offset = cell.getFamilyOffset();
				length = cell.getFamilyLength();
				datas = new String(data, offset, length, charset);
				
				matcher = datas.matches(regEx);
				if (!matcher) {
					for (int i = 0; i < data.length; i++) {
				//		System.out.print(data[i]);
					}
				//	System.out.print(offset);
				//	System.out.println(length);
					System.out.println(datas);
				}

				data = cell.getQualifierArray();
				offset = cell.getQualifierOffset();
				length = cell.getQualifierLength();
				datas = new String(data, offset, length, charset);
				
				matcher = datas.matches(regEx);
				if (!matcher) {
					for (int i = 0; i < data.length; i++) {
		//				System.out.print(data[i]);
					}
					//System.out.print(offset);
					//System.out.println(length);
					System.out.println(datas);
				}

				data = cell.getValueArray();
				offset = cell.getValueOffset();
				length = cell.getValueLength();
				datas = new String(data, offset, length, charset);
				
				matcher = datas.matches(regEx);
				if (!matcher) {
					for (int i = 0; i < data.length; i++) {
				//		System.out.print(data[i]);
					}
				//	System.out.print(offset);
				//	System.out.println(length);
					System.out.println(datas);
				}

			}
		}
		htable.close();
	}

	public static String hbase2url(String url) {

		String hostZA = url.substring(0, url.indexOf(":"));
		url = url.substring(url.indexOf(":") + 1);
		int end = url.lastIndexOf("/") == url.length() - 1 ? url.length() - 1
				: url.length();
		String path = url.substring(url.indexOf("/"), end);
		url = url.substring(0, url.indexOf("/")) + "://";
		String[] hosts = hostZA.split("\\.");
		for (int i = hosts.length - 1; i > 0; i--) {
			url += (hosts[i] + ".");
		}
		return url += (hosts[0] + path);
	}

	public static void createTable(String tableName, String familys[])
			throws Exception {

		HBaseAdmin admin = new HBaseAdmin(config); // 客户端管理工具类
		if (admin.tableExists(tableName)) {
			admin.disableTable(tableName);
			admin.deleteTable(tableName);
			System.out.println("delete table " + tableName + " ok.");
		}
		HTableDescriptor tableDesc = new HTableDescriptor(
				TableName.valueOf(tableName));
		for (int i = 0; i < familys.length; i++) {
			tableDesc.addFamily(new HColumnDescriptor(familys[i]));
		}
		admin.createTable(tableDesc);
		admin.close();
		System.out.println("create table " + tableName + " ok.");
	}

	public static Elements getElements(Result value, String cssQuery) {
		byte[] cnts = value.getValue(Bytes.toBytes("f"), Bytes.toBytes("cnt"));
		if (cnts == null || cnts.length == 0)
			return null;
		String cnt = null;
		try {
			cnt = new String(cnts, "GB18030");
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
		}
		if (cnt == null || cnt.equals(""))
			return null;
		Document doc = Jsoup.parse(cnt);
		Elements elements = doc.select(cssQuery);
		return elements;
	}

	public static class MyMapper extends TableMapper<Text, Text> {

		public void map(ImmutableBytesWritable row, Result value,
				Context context) throws IOException, InterruptedException {
			String url = Bytes.toString(row.get());
			url = hbase2url(url);

			Elements bookNames = getElements(value,
					"div.keywordthis span.red13");
			if (bookNames == null || bookNames.size() == 0)
				return;
			String bookNameText = bookNames.text();
			if (bookNameText == null || bookNameText.equals(""))
				return;
			if (bookNames.size() > 1)
				bs.add(bookNameText);
			bookNameText = bookNameText.replaceAll("^ *| *$", "");
			bookNameText = bookNameText.trim();
			if (bookNameText == null || bookNameText.equals(""))
				return;
			context.write(new Text(bookNameText), new Text("bookschina：" + url));
			++books;
			Elements bookInfos = getElements(value, "td.wordheight");
			for (Element bookInfo : bookInfos) {
				String bookInfoKey = bookInfo.text();
				if (bookInfoKey == null || bookInfoKey.equals(""))
					continue;
				String bookInfoValue = "";
				if (bookInfoKey.equals("出 版 社：")) {
					bookInfoValue = bookInfo.nextElementSibling().child(0)
							.text();
				} else {
					bookInfoValue = bookInfo.nextElementSibling().text();
				}
				if (bookInfoValue == null || bookInfoValue.equals(""))
					continue;
				context.write(new Text(bookNameText), new Text(bookInfoKey
						+ bookInfoValue));
			}

		}
	}

	public static class MyTableReducer extends
			TableReducer<Text, Text, ImmutableBytesWritable> {
		public void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			Put put = new Put(Bytes.toBytes(key.toString()));
			int cou = -1;
			List<String> urls = new ArrayList<String>();
			for (Text val : values) {
				String[] vals = val.toString().split("：");
				if (vals.length > 2) {
					for (int i = 2; i < vals.length; i++) {
						vals[1] += "：" + vals[i];
					}
					inf.add("\n" + val.toString());
					inf.add("\n" + vals[0] + "：" + vals[1]);
				}
				put.add(Bytes.toBytes("info"), Bytes.toBytes(vals[0]),
						Bytes.toBytes(vals[1]));
				if (vals[0].equals("bookschina")) {
					++cou;
					urls.add("\n" + vals[1]);
				}
			}
			context.write(null, put);
			if (cou > 0) {
				co += cou;
				url.addAll(urls);
			}
		}
	}
}